package com.millstein.realtime.app.dwd.db;

import com.millstein.realtime.app.base.BaseSqlApp;
import com.millstein.realtime.common.Constants;
import com.millstein.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Description
 * @Author tsing
 * @Date 2024-10-14 14:27
 */
public class Dwd_14_InteractionFavorAdd extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_14_InteractionFavorAdd().init(
                7008,
                3,
                "Dwd_14_InteractionFavorAdd"
        );
    }

    /**
     * 具体数据处理的逻辑，由子类编写
     *
     * @param env      执行环境
     * @param tableEnv 表执行环境
     */
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1.读取ods层的数据
        readOdsDataFromKafka(tableEnv, "Dwd_14_InteractionFavorAdd");

        // 2.过滤出收藏的数据
        Table favorInfo = tableEnv.sqlQuery(
                "select " +
                "    `data`['id'] id, " +
                "    `data`['user_id'] user_id, " +
                "    `data`['sku_id'] sku_id, " +
                "    date_format(`data`['create_time'], 'yyyy-MM-dd') date_id, " +
                "    `data`['create_time'] create_time, " +
                "    `ts` " +
                "from maxwell_table " +
                "where `database` = 'gmall' " +
                "    and `table` = 'favor_info' " +
                "    and `type` = 'insert'"
        );

        // 3.创建kafka-sink动态表
        tableEnv.executeSql(
                "create table dwd_interaction_favor_add ( " +
                "    id string, " +
                "    user_id string, " +
                "    sku_id string, " +
                "    date_id string, " +
                "    create_time string, " +
                "    ts bigint " +
                ")" + SqlUtil.getKafkaSinkDDL(Constants.TOPIC_DWD_INTERACTION_FAVOR_ADD)
        );

        // 4.处理后的数据写到动态表中
        favorInfo.executeInsert("dwd_interaction_favor_add");
    }
}
